package org.example;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.FileUtils;

import java.io.File;

public class FlinkSqlKafkaToMysql {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 读取SQL文件内容
        String sqlContent = FileUtils.readFile(new File(FlinkSqlKafkaToMysql.class.getResource("/kafka_to_oracle.sql").getFile()),"UTF-8");

        // 分割SQL语句
        String[] sqlStatements = sqlContent.split(";");

        // 执行每条SQL语句
        for (String sql : sqlStatements) {
            sql = sql.trim();
            if (!sql.isEmpty()) {
                System.out.println("Executing SQL: " + sql);
                tableEnv.executeSql(sql);
            }
        }

        // 执行环境
        env.execute("FlinkSQL Kafka to Mariadb");

    }

}    